/**
 * Copyright 2010 - 2017 JetBrains s.r.o.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package jetbrains.exodus.entitystore;

import jetbrains.exodus.backup.Backupable;
import jetbrains.exodus.core.dataStructures.hash.LongHashMap;
import jetbrains.exodus.core.dataStructures.hash.LongSet;
import jetbrains.exodus.env.Transaction;
import jetbrains.exodus.util.*;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;

/**
 * {@code BlobVault} is base class describing interface to <a href="https://en.wikipedia.org/wiki/Binary_large_object">
 * binary large objects (BLOBs)</a> used internally by implementation of {@linkplain PersistentEntityStore}.
 *
 * <p>Blob values are identified by blob <i>handles</i> generated by {@linkplain BlobHandleGenerator}.
 * {@code PersistentEntityStore} encapsulates mapping of {@linkplain Entity} blob values to blob handles, whereas
 * {@code BlobVault} allows to read/write blob values identified by blob handles. Blob values are immutable, i.e.
 * blob value by a blob handle cannot be changed, only deleted.
 *
 * <p>{@code BlobVault} allows to get string content of blob for method {@linkplain Entity#getBlobString(String)}.
 * In order to reduce CPU consumption on transformation of binary content to UTF-8 strings, {@code BlobVault}
 * encapsulates a cache of blob strings.
 *
 * @see PersistentEntityStore
 * @see BlobHandleGenerator
 */
public abstract class BlobVault implements BlobHandleGenerator, Backupable {

    private static final int READ_BUFFER_SIZE = 0x4000;
    static final ByteArraySpinAllocator bufferAllocator = new ByteArraySpinAllocator(READ_BUFFER_SIZE);
    private static final BlobStringsCache.BlobStringsCacheCreator
        stringContentCacheCreator = new BlobStringsCache.BlobStringsCacheCreator();
    private static final IdGenerator identityGenerator = new IdGenerator();

    private final PersistentEntityStoreConfig config;
    private final BlobStringsCache stringContentCache;
    private final int vaultIdentity;

    protected BlobVault(@NotNull final PersistentEntityStoreConfig config) {
        this.config = config;
        stringContentCache = config.isBlobStringsCacheShared() ?
            stringContentCacheCreator.getInstance() :
            new BlobStringsCache.BlobStringsCacheCreator().getInstance();
        vaultIdentity = identityGenerator.nextId();
    }

    public int getIdentity() {
        return vaultIdentity;
    }

    /**
     * Returns binary content of blob identified by specified blob handle as {@linkplain InputStream}.
     *
     * @param blobHandle blob handle
     * @param txn        {@linkplain Transaction} instance
     * @return binary content of blob as {@linkplain InputStream}
     * @see Entity#getBlob(String)
     */
    @Nullable
    public abstract InputStream getContent(final long blobHandle, @NotNull final Transaction txn);

    /**
     * Returns size of blob identified by specified blob handle in bytes.
     *
     * @param blobHandle blob handle
     * @param txn        {@linkplain Transaction} instance
     * @return size of blob in bytes
     * @see Entity#getBlobSize(String)
     */
    public abstract long getSize(final long blobHandle, @NotNull final Transaction txn);

    /**
     * Returns {@code true} if {@code BlobVault} implementation requires {@linkplain Transaction transaction} to
     * flush blobs when {@linkplain StoreTransaction#flush()} or {@linkplain StoreTransaction#commit()} is called.
     *
     * @return {@code true} if {@code BlobVault} implementation requires {@linkplain Transaction transaction}
     */
    public abstract boolean requiresTxn();

    /**
     * Method called by {@linkplain PersistentEntityStore} to flush blobs identified by blobs. Two maps represent two
     * ways of dealing with blob content: using streams and {@code java.io.File} instances. They naturally follow from
     * the {@linkplain Entity} API, having two methods to set blob: {@linkplain Entity#setBlob(String, InputStream)}
     * and {@linkplain Entity#setBlob(String, File)}.
     *
     * <p>The set {@code deferredBlobsToDelete} contains blob handles of blobs that should be deleted after the
     * transaction is finished. In most cases, these blobs cannot be deleted immediately without violating isolation
     * transaction guarantees.
     *
     * <p>Although transaction passed to the method is {@code @NotNull} it can be already finished (check by
     * {@linkplain Transaction#isFinished()}). E.g., {@linkplain StoreTransaction#commit()} at first commits
     * underlying environment {@linkplain Transaction} and then flushes blobs passing instance of committed transaction.
     *
     * @param blobStreams           map of blob handles to {@linkplain InputStream} instances
     * @param blobFiles             map of blob handles to {@linkplain File} instances
     * @param deferredBlobsToDelete set of blob handles of blobs that should be deleted after the transaction is finished
     * @param txn                   {@linkplain Transaction} instance
     * @throws Exception something went wrong
     * @see LongHashMap
     * @see LongSet
     * @see Transaction
     */
    public abstract void flushBlobs(@Nullable final LongHashMap<InputStream> blobStreams,
                                    @Nullable final LongHashMap<File> blobFiles,
                                    @Nullable final LongSet deferredBlobsToDelete,
                                    @NotNull final Transaction txn) throws Exception;

    /**
     * Returns amount of disk space occupied by the {@code BlobVault}. This value is rather "rough" since it does not
     * take into consideration particular file system features of storage device.
     *
     * @return amount of disk space occupied by the {@code BlobVault}
     * @see PersistentEntityStore#getUsableSpace()
     * @see IOUtil#getAdjustedFileLength(File)
     */
    public abstract long size();

    /**
     * Is called on {@linkplain PersistentEntityStore#clear()}.
     */
    public abstract void clear();

    /**
     * Is called on {@linkplain PersistentEntityStore#close()}.
     */
    public abstract void close();

    /**
     * Returns string content of blob identified by specified blob handle. String contents cache is used.
     *
     * @param blobHandle blob handle
     * @param txn        {@linkplain Transaction} instance
     * @return string content of blob identified by specified blob handle
     * @throws IOException
     */
    @Nullable
    public final String getStringContent(final long blobHandle, @NotNull final Transaction txn) throws IOException {
        String result;
        result = stringContentCache.tryKey(this, blobHandle);
        if (result == null) {
            final InputStream content = getContent(blobHandle, txn);
            result = content == null ? null : UTFUtil.readUTF(content);
            if (result != null && result.length() <= config.getBlobStringsCacheMaxValueSize()) {
                if (stringContentCache.getObject(this, blobHandle) == null) {
                    stringContentCache.cacheObject(this, blobHandle, result);
                }
            }
        }
        return result;
    }

    /**
     * @return {@linkplain jetbrains.exodus.core.dataStructures.CacheHitRateable#hitRate() hit rate} of string
     * contents cache
     */
    public final float getStringContentCacheHitRate() {
        return stringContentCache.getHitRate();
    }

    public final ByteArrayOutputStream copyStream(@NotNull final InputStream source,
                                                  final boolean closeSource) throws IOException {
        final ByteArrayOutputStream memCopy = new LightByteArrayOutputStream();
        IOUtil.copyStreams(source, memCopy, bufferAllocator);
        if (closeSource) {
            source.close();
        }
        return memCopy;
    }

    public final ByteArraySizedInputStream cloneStream(@NotNull final InputStream source,
                                                       final boolean closeSource) throws IOException {
        final ByteArrayOutputStream memCopy = copyStream(source, closeSource);
        return new ByteArraySizedInputStream(memCopy.toByteArray(), 0, memCopy.size());
    }
}
